package com.haiyou.data.common.persister.mysql.persister.delay;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.springframework.util.ObjectUtils;

import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;
import com.haiyou.common.spring.schedule.ActorScheduledExecutor;
import com.haiyou.common.thread.ThreadManager;
import com.haiyou.common.thread.ThreadManager.ThreadEnum;
import com.haiyou.common.thread.policy.ShutdownFlushRejectedPolicy;
import com.haiyou.common.thread.task.AbstractOrderedTask;
import com.haiyou.common.util.random.ProbabilityUtils;
import com.haiyou.common.util.reflect.AnnotationFinder;
import com.haiyou.common.util.string.StringFormatUitls;
import com.haiyou.data.common.persister.mysql.annotation.SyncPolicy;
import com.haiyou.data.common.persister.mysql.annotation.Table.RollPolicy;
import com.haiyou.data.common.persister.mysql.help.TableHelper;
import com.haiyou.data.common.persister.mysql.persister.PreparedPersister;
import com.haiyou.data.common.persister.mysql.persister.condition.QueryConditions;
import com.haiyou.data.common.persister.mysql.persister.page.PagePreparedPersister;

import ch.qos.logback.core.rolling.RollingPolicy;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 只能在单点服务使用，不可以用于分布式! 延迟持久器,直接new出来一个实例，各种set完之后调用save，通常用于日志打点系统
 * 
 * @author xingyuan
 *
 * @param <E>
 */
@Slf4j
public abstract class DelayQueuePreparedPersister
		<Id extends Serializable & Comparable<Id>, E extends DelayEntity<Id>>
		extends PagePreparedPersister<Id, E> {

	private static final class Lazy {
		private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE;
		static {
			int nThreads = ThreadManager.getThreadCount(ThreadEnum.delay_persister_thread);
			SCHEDULED_EXECUTOR_SERVICE = new ActorScheduledExecutor(nThreads,
					DelayQueuePreparedPersister.class.getSimpleName(),ShutdownFlushRejectedPolicy.instance);
		}
	}

	public static ScheduledExecutorService scheduler() {
		return Lazy.SCHEDULED_EXECUTOR_SERVICE;
	}

	private Queue<E> queue = Queues.newLinkedBlockingQueue();

	private int initialDelay = 60;
	private int delay = 60;
	private TimeUnit timeUnit = TimeUnit.SECONDS;

	private ScheduledFuture<?> scheduledFuture;

	@Override
	public void init() {
		super.init();
		SyncPolicy syncPolicy = AnnotationFinder.findAnnotation(getClass(), SyncPolicy.class);
		if (syncPolicy != null) {
			initialDelay = syncPolicy.initialDelay();
			delay = syncPolicy.delay();
			timeUnit = syncPolicy.timeUnit();
		}
		if (delay > 0) {
			initialDelay = initialDelay > 0 ? ProbabilityUtils.xTo2x(initialDelay) : 0;
			delay = ProbabilityUtils.xTo2x(delay);
			delay = Math.max(1, delay);
			scheduledFuture = scheduler().scheduleAtFixedRate(new DelayTask(false), initialDelay, delay, timeUnit);
			log.info("{} scheduled, initialDelay={},delay={}", this, initialDelay, delay);
		} else {
			log.info("{} no scheduled, save now async", this);
		}
	}

	@Override
	public void destory() {
		super.destory();
		//log.info("==========={} 关闭服务器执行==========",this.getClass().getSimpleName());
		//scheduler().execute(new DelayTask(true));
		try {
			//保证 数据正常刷入 这里仅日志业务需要
			new DelayTask(true).exec();
		} catch (Exception e) {
			log.error(e.getMessage(),e);
		}
		//log.info("=========执行完成===========");
	}

	public void flushAll() {
		scheduler().execute(new DelayTask(false));
	}
	
	/**
	 * 直接插入
	 * 
	 * @param log
	 */
	public void save(E log) {
		if (Objects.isNull(log.getId())) {
			if (!TableHelper.pkAutoIncr(tableMeta)) {
				throw new RuntimeException("非自增主键必须设置主键的值:" + log);
			}
		}
		if (Objects.nonNull(log.getId())) {
			if (TableHelper.pkAutoIncr(tableMeta)) {
				throw new RuntimeException("自增主键不需要设置主键的值:" + log);
			}
		}
		if (queue.offer(log)) {
			if (scheduledFuture == null || sqlBatchNum() <= queue.size()) {
				scheduler().execute(new DelayTask(false));
			}
		}
	}

	private final class DelayTask extends AbstractOrderedTask<Class<E>> {

		private final boolean last;

		private DelayTask(boolean last) {
			super(entityClass);
			this.last = last;
		}
		
		@Override
		protected void exec() throws Exception {
			if (queue.isEmpty())
				return;
			Set<E> saveValues = null;
			while (!queue.isEmpty()) {
				if (saveValues == null) {
					saveValues = Sets.newLinkedHashSet();
				}
				E value = queue.poll();
				saveValues.add(value);
				if (saveValues.size() >= sqlBatchNum() || queue.isEmpty()) {
					insert(saveValues);
					saveValues = null;
				}
			}
			if (last && !queue.isEmpty())
				exec();
		}
	}
	
	/**
	 * 业务逻辑请自己排序
	 */
	@Override
	public List<E> queryPageByCondition(int page,int limit){
		return queryPageByCondition(page, limit, new QueryConditions());
	}
	
	/**
	 * 分表分页逻辑
	 */
	@Override
	public List<E> queryPageByCondition(int page,int limit,QueryConditions condition){
		
		//只有一个db 目前版本
		String db = databaseMeta.getReallyNames().iterator().next();
	
		List<String> tableNames = Lists.newArrayList(TableHelper.allTables(tableMeta));
	
		//分表方式
		RollPolicy rollingPolicy = tableMeta.getRollPolicy();
		
		if(rollingPolicy == RollPolicy.NONE) {
			//单表结构
			return super.queryPageByCondition(page, limit, condition);
		}
		
		//针对表名进行排序
		tableNames = sortTableNames(tableNames);
		
		//page 页数 从1
		//开始limit 单页数量
		int prevNum = (page - 1) * limit; // 之前页面条数
		// 所有符合条件的总条数
		int nowNum = 0;
		int i = 0;
		Map<String, DelayPageParam> params = Maps.newHashMap();
		//表数量
		int arrNum = tableNames.size();
		
		for (String tableName : tableNames) {
			
			//这个表有多少条数据
			int countNum = (int)super.queryCountByCondition(db, tableName, condition);
			//当前总条数
			nowNum += countNum;
			// 大于上页数据则为当前页需要显示数据
			int pageSize = 0;
			
			if (nowNum > prevNum) {
				int offset = 0;
				
				if (params.size() > 0) {
					int alyNum = 0;
					for (Entry<String, DelayPageParam> entry : params.entrySet()) {
						DelayPageParam param = entry.getValue();
						alyNum +=  param.getPageSize();
					}

					offset = countNum - (nowNum - prevNum) + alyNum;
					pageSize = ((limit - alyNum) < countNum) ? (limit - alyNum) : countNum;
				} else {
					offset = countNum - (nowNum - prevNum);
					pageSize = (nowNum - prevNum > limit) ? limit : (nowNum - prevNum);
				}
			
				if(offset < 0) {
					break;
				}
				
				params.put(tableName, DelayPageParam.build(offset, pageSize));
			}
			
			i++; // 满足最后一页查询不足10条
			// 足够当前页显示则返回
			if (pageSize == limit || i == arrNum) {
				break;
			}
		}
		
		List<E> lists = Lists.newArrayList();

		for (String tableName : tableNames) {
			
			//log.info("分页结果 : {} {}",tableName,JSONObject.toJSONString(params.getOrDefault(tableName, null)));
		
			DelayPageParam param = params.getOrDefault(tableName, null);
			
			if(param == null) {
				continue;
			}
			//查询数据
			List<E> list = super.queryPageByOffsetAndCondition(db, tableName, param.getOffset(), param.getPageSize(),pageSqlDesc, condition);
			
			if(!ObjectUtils.isEmpty(list)) {
				lists.addAll(list);
			}
		}
		
		return lists;
	}

	
	public static void main(String[] args) {
		
		int page = 2;
		int limit = 11;
		
		List<String> tableNames = Lists.newArrayList("1","2","3","4");

		Map<String, Integer> tableCounts = Maps.newHashMap();
		
		tableCounts.put("1", 10);
		tableCounts.put("2", 10);
		tableCounts.put("3", 10);
		tableCounts.put("4", 10);

		//分页查询参数
		
		int prevNum = (page - 1) * limit; // 之前页面条数
		int nowNum = 0;
		int i = 0;
		Map<String, DelayPageParam> params = Maps.newHashMap();
		//表数量
		int arrNum = tableCounts.size();
		
		for (String tableName : tableNames) {
			
			//这个表有多少个
			int countNum = tableCounts.get(tableName);
			//当前总条数
			nowNum += countNum;
			// 大于上页数据则为当前页需要显示数据
			int pageSize = 0;
			
			if (nowNum > prevNum) {
				int offset = 0;
				
				if (params.size() > 0) {
					int alyNum = 0;
					for (Entry<String, DelayPageParam> entry : params.entrySet()) {
						DelayPageParam param = entry.getValue();
						alyNum +=  param.getPageSize();
					}

					offset = countNum - (nowNum - prevNum) + alyNum;
					pageSize = ((limit - alyNum) < countNum) ? (limit - alyNum) : countNum;
				} else {
					offset = countNum - (nowNum - prevNum);
					pageSize = (nowNum - prevNum > limit) ? limit : (nowNum - prevNum);
				}
			
				if(offset < 0) {
					break;
				}
				
				params.put(tableName, DelayPageParam.build(offset, pageSize));
			}
			
			i++; // 满足最后一页查询不足10条
			// 足够当前页显示则返回
			if (pageSize == limit || i == arrNum) {
				break;
			}
		}
		
		
		log.info("=========================");
		
		for (String tableName : tableNames) {
			log.info("分页结果 : {} {}",tableName,JSONObject.toJSONString(params.getOrDefault(tableName, null)));
		}
	
		
	}
	
	
	
	
	
}
